package com.aioething.iot.hook;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.aioething.iot.dao.EquipmentDao;
import com.aioething.iot.dao.OtaTaskDao;
import com.aioething.iot.model.Equipment;
import com.aioething.iot.model.OtaTask;
import com.aioething.iot.service.IEmqxService;
import io.emqx.exhook.*;
import io.grpc.stub.StreamObserver;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.tio.utils.lock.MapWithLock;

import java.io.IOException;
import java.util.Arrays;
import java.util.Date;

@Component
public class HookProvider extends HookProviderGrpc.HookProviderImplBase {
    private static final Logger logger = LogManager.getLogger(HookProvider.class);

    @Autowired
    private EquipmentDao equipmentDao;
    @Autowired
    private OtaTaskDao otaTaskDao;
    @Autowired
    private IEmqxService emqxService;

    @Override
    public void onProviderLoaded(ProviderLoadedRequest request, StreamObserver<LoadedResponse> responseObserver) {
        logger.debug("onProviderLoaded");
        HookSpec[] specs = {
                HookSpec.newBuilder().setName("client.connected").build(),
                HookSpec.newBuilder().setName("client.disconnected").build(),
                HookSpec.newBuilder().setName("message.publish")
                        .addTopics("things/+/+/property/up")
                        .addTopics("things/+/+/ota/up")
                        .addTopics("things/+/+/action/up")
                        .build()
        };
        LoadedResponse reply = LoadedResponse.newBuilder().addAllHooks(Arrays.asList(specs)).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onProviderUnloaded(ProviderUnloadedRequest request, StreamObserver<EmptySuccess> responseObserver) {
        logger.debug("onProviderUnloaded");
        EmptySuccess reply = EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onClientConnected(ClientConnectedRequest request, StreamObserver<EmptySuccess> responseObserver) {
        EmptySuccess reply = EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
        logger.debug("客户端[{}]上线", request.getClientinfo().getClientid());
        try {
            Equipment equipment = new Equipment();
            equipment.setProductId(request.getClientinfo().getUsername());
            equipment.setDeviceName(request.getClientinfo().getClientid());
            equipment.setStatus(1);
            equipment.setOnlineTime(new Date());
            equipmentDao.updateByPrimaryKeySelective(equipment);
        } catch (Exception e) {
            logger.error(e);
        }
    }

    @Override
    public void onClientDisconnected(ClientDisconnectedRequest request, StreamObserver<EmptySuccess> responseObserver) {
        EmptySuccess reply = EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
        logger.debug("客户端[{}]离线", request.getClientinfo().getClientid());
        try {
            Equipment equipment = new Equipment();
            equipment.setProductId(request.getClientinfo().getUsername());
            equipment.setDeviceName(request.getClientinfo().getClientid());
            equipment.setStatus(0);
            equipment.setOfflineTime(new Date());
            equipmentDao.updateByPrimaryKeySelective(equipment);
        } catch (Exception e) {
            logger.error(e);
        }
    }

    @Override
    public void onMessagePublish(MessagePublishRequest request, StreamObserver<ValuedResponse> responseObserver) {
        ValuedResponse reply = ValuedResponse.newBuilder().setType(ValuedResponse.ResponsedType.CONTINUE).setMessage(request.getMessage()).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
        String topic = request.getMessage().getTopic();
        String items[] = topic.split("/");
        JSONObject payload = JSONUtil.parseObj(request.getMessage().getPayload().toStringUtf8());
        logger.info(String.format("topic:%s, payload:%s", topic, payload));
        if (items.length < 5)   return;
        if ("property".equals(items[3]) && "up".equals(items[4])) {
            // 物模型上报
            String equipmentId = items[1].concat("/").concat(items[2]);
            Equipment exists = equipmentDao.selectByPrimaryKey(equipmentId);
            if (ObjectUtil.isNull(exists))  return;
            Equipment equipment = new Equipment();
            equipment.setEquipmentId(equipmentId);
            JSONObject reported = exists.getReported();
            JSONObject params = payload.getJSONObject("params");
            if (params.containsKey("reported")) {
                for (String key : params.getJSONObject("reported").keySet()) {
                    reported.set(key, params.getJSONObject("reported").get(key));
                    if ("version".equals(key)) {
                        equipment.setVersion(params.getJSONObject("reported").getStr(key));
                    } else if ("iccid".equals(key)) {
                        equipment.setIccid(params.getJSONObject("reported").getStr(key));
                    }
                }
            }
            equipment.setReported(reported);
            JSONObject desired = exists.getDesired();
            if (params.containsKey("desired")) {
                for (String key : params.getJSONObject("desired").keySet()) {
                    desired.set(key, params.getJSONObject("desired").get(key));
                }
            }
            equipment.setDesired(desired);
            // 计算差异
            JSONObject delta = exists.getDelta();
            for (String key : desired.keySet()) {
                if (ObjectUtil.notEqual(desired.get(key), reported.get(key))) {
                    delta.set(key, desired.get(key));
                } else {
                    delta.remove(key);
                }
            }
            equipment.setDelta(delta);
            equipmentDao.updateByPrimaryKeySelective(equipment);
            try {
                emqxService.publish(String.format("things/%s/%s/property/down", items[1], items[2]),
                        JSONUtil.createObj().set("method", "report_reply").set("clientToken", payload.getStr("clientToken")).toString());
            } catch (IOException e) {
                logger.error(e.getMessage());
            }
            if (!delta.isEmpty()) {
                try {
                    emqxService.publish(String.format("things/%s/%s/property/down", items[1], items[2]),
                            JSONUtil.createObj().set("method", "delta").set("params", delta).toString());
                } catch (IOException e) {
                    logger.error(e.getMessage());
                }
            }
        } else if ("ota".equals(items[3]) && ("up".equals(items[4]))) {
            String method = payload.getStr("method");
            if ("ota_reply".equals(method)) {
                String clientToken = payload.getStr("clientToken");
                // 同步调用响应
                MapWithLock<String, String> syncLock = emqxService.getSyncLock();
                String lockObj = syncLock.get(clientToken);
                if (ObjectUtil.isNotNull(lockObj)) {
                    synchronized (lockObj) {
                        syncLock.put(clientToken, payload.getJSONObject("params").toString());
                        lockObj.notify();
                    }
                }
            } else if ("ota_progress".equals(method)) {
                JSONObject params = payload.getJSONObject("params");
                OtaTask otaTask = new OtaTask();
                otaTask.setTaskId(params.getStr("taskId"));
                otaTask.setStatus(params.getInt("status"));
                otaTask.setProgress(params.getInt("progress"));
                otaTaskDao.updateByPrimaryKeySelective(otaTask);
            }
        } else if ("ota".equals(items[3]) && ("up".equals(items[4]))) {
            String clientToken = payload.getStr("clientToken");
            // 同步调用响应
            MapWithLock<String, String> syncLock = emqxService.getSyncLock();
            String lockObj = syncLock.get(clientToken);
            if (ObjectUtil.isNotNull(lockObj)) {
                synchronized (lockObj) {
                    syncLock.put(clientToken, payload.getJSONObject("params").toString());
                    lockObj.notify();
                }
            }
        }
    }
}
